package avicit.bdp.dgs.metadata.compare.util;

import avicit.bdp.common.service.dto.BdpColumnDTO;
import avicit.bdp.common.service.dto.BdpTableDTO;
import avicit.bdp.dgs.metadata.compare.pojo.MetadataTableChange;
import avicit.bdp.dgs.metadata.compare.pojo.MetadataTableColumnDiffItem;
import avicit.bdp.dgs.metadata.compare.pojo.MetadataTableDiffItem;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.commons.collections4.CollectionUtils;

/**
 * 元数据数据表比对工具类
 *
 * @author mayi
 */
public class MetadataCompareUtil {

  public static List<MetadataTableChange> batchCompare(
      List<BdpTableDTO> oldTables, List<BdpTableDTO> newTables) {

    List<MetadataTableChange> result = new ArrayList<>();

    if (oldTables == null) {
      oldTables = new ArrayList<>();
    }

    if (newTables == null) {
      newTables = new ArrayList<>();
    }

    final Map<String, BdpTableDTO> newTablesMap =
        newTables.stream().collect(Collectors.toMap(BdpTableDTO::getName, table -> table));

    final Map<String, BdpTableDTO> oldTablesMap =
        oldTables.stream().collect(Collectors.toMap(BdpTableDTO::getName, table -> table));

    final Set<String> newTableNameSet = newTablesMap.keySet();
    final Set<String> oldTableNameSet = oldTablesMap.keySet();

    // 新表集合中存在，旧表集合中不存在，表示新增
    Set<String> addTableNameSet =
        newTableNameSet.stream()
            .filter(e -> !oldTableNameSet.contains(e))
            .collect(Collectors.toSet());

    if (CollectionUtils.isNotEmpty(addTableNameSet)) {
      List<MetadataTableChange> addTableChange =
          addTableNameSet.stream()
              .map(
                  r -> {
                    MetadataTableChange change = new MetadataTableChange(newTablesMap.get(r));
                    change.setTableDiff(new MetadataTableDiffItem(null, newTablesMap.get(r)));
                    return change;
                  })
              .collect(Collectors.toList());
      result.addAll(addTableChange);
    }

    // 旧表集合中存在，新表集合中不存在，表示删除
    Set<String> dropTableNameSet =
        oldTableNameSet.stream()
            .filter(e -> !newTableNameSet.contains(e))
            .collect(Collectors.toSet());

    if (CollectionUtils.isNotEmpty(dropTableNameSet)) {
      List<MetadataTableChange> addTableChange =
          dropTableNameSet.stream()
              .map(
                  r -> {
                    MetadataTableChange change = new MetadataTableChange(oldTablesMap.get(r));
                    change.setTableDiff(new MetadataTableDiffItem(oldTablesMap.get(r), null));
                    return change;
                  })
              .collect(Collectors.toList());
      result.addAll(addTableChange);
    }

    // 新旧集合中均存在，查找变更的数据集合
    Set<String> alertTableNameSet =
        oldTableNameSet.stream().filter(newTableNameSet::contains).collect(Collectors.toSet());

    if (CollectionUtils.isNotEmpty(alertTableNameSet)) {
      List<MetadataTableChange> alertChanges =
          alertTableNameSet.stream()
              .map(
                  alertTableName ->
                      buildAlterTableChange(
                          newTablesMap.get(alertTableName), oldTablesMap.get(alertTableName)))
              .collect(Collectors.toList());
      result.addAll(alertChanges);
    }

    return result;
  }

  private static MetadataTableChange buildAlterTableChange(
      BdpTableDTO newTable, BdpTableDTO oldTable) {
    MetadataTableDiffItem tableDiff = new MetadataTableDiffItem(newTable, oldTable);
    List<MetadataTableColumnDiffItem> columnDiff = new ArrayList<>();

    boolean tableChanged = false;

    for (BdpColumnDTO oldColumn : oldTable.getColumnList()) {
      Optional<BdpColumnDTO> col =
          newTable.getColumnList().stream()
              .filter(t -> t.getName().equals(oldColumn.getName()))
              .findFirst();

      MetadataTableColumnDiffItem item =
          new MetadataTableColumnDiffItem(col.orElse(null), oldColumn);
      if (!columnDiff.contains(item)) {
        columnDiff.add(item);
        if (!MetadataTableColumnDiffItem.NO.equals(item.getType())) {
          tableChanged = true;
        }
      }
    }

    for (BdpColumnDTO r : newTable.getColumnList()) {
      Optional<BdpColumnDTO> col =
          oldTable.getColumnList().stream()
              .filter(t -> t.getName().equals(r.getName()))
              .findFirst();

      MetadataTableColumnDiffItem item = new MetadataTableColumnDiffItem(r, col.orElse(null));
      if (!MetadataTableColumnDiffItem.NO.equals(item.getType())) {
        if (!columnDiff.contains(item)) {
          columnDiff.add(item);
          if (!MetadataTableColumnDiffItem.NO.equals(item.getType())) {
            tableChanged = true;
          }
        }
      }
    }

    MetadataTableChange change = new MetadataTableChange(oldTable);
    change.setTableDiff(tableDiff);
    change.setColumnDiffs(columnDiff);
    if (tableChanged) {
      tableDiff.setType(MetadataTableDiffItem.ALTER);
    }
    return change;
  }
}
